package com.bigdata.topN

import org.apache.spark.sql.SparkSession

import scala.collection.mutable.ListBuffer

/**
  * 文章TopN 分析
  * 计算方法： 浏览量*1 + 评论数*2 + 点赞数*3
  */
object ArticleTopN {
  def main(args: Array[String]): Unit = {
    val session = SparkSession.builder().appName("articleTopN").master("local").getOrCreate()
    session.sqlContext.read.load(Constants.HADOOP_STORAGE+Constants.TIME_GENERATOR+"-"+Constants.TABLE_ARTICLE)
      .createOrReplaceTempView("article")
    session.sqlContext.read.load(Constants.HADOOP_STORAGE+Constants.TIME_GENERATOR+"-"+Constants.TABLE_BROWSE)
      .createOrReplaceTempView("browse")
    session.sqlContext.read.load(Constants.HADOOP_STORAGE+Constants.TIME_GENERATOR+"-"+Constants.TABLE_COMMENT)
      .createOrReplaceTempView("comment")
    session.sqlContext.read.load(Constants.HADOOP_STORAGE+Constants.TIME_GENERATOR+"-"+Constants.TABLE_SUPPORT)
      .createOrReplaceTempView("support")

    session.sql("show tables").show()
//    session.sql("select * from article").show()
//    session.sql("select * from support").show()
//    session.sql("select * from article a, browse b where a.id=b.article_id").show()
    // 拿到文章id 和 点赞的个数
    val supportCountDF = session.sql("select a.id,count(a.id) as count from article a, support s where a.id=s.article_id group by a.id")
    println("=================文章id 和 点赞数=====================")
    println(supportCountDF.show())
    supportCountDF.createOrReplaceTempView("supportCount")

    // 拿到文章id 和 评论个数
    val commentCountDF = session.sql("select a.id,count(a.id) as count from article a,comment as c where a.id = c.article_id group by a.id ")
    println("==================评论数=================")
    commentCountDF.show()
    commentCountDF.createOrReplaceTempView("commentCount")

    // 计算并生成新的 df
    val resultDF = session.sql("select s.id,s.count*3+c.count*2 as weight from supportCount s join commentCount c on s.id=c.id order by weight desc limit 15")
    resultDF.show()

    // 保存到数据库
    resultDF.foreachPartition(partition => {
      val list = new ListBuffer[Article]
      partition.foreach(info => {
        val articleId = info.getAs[String]("id")
        val weight = info.getAs[Long]("weight")

        list.append(Article(articleId,weight,Constants.TIME_GENERATOR))
      })
      ArticleDAO.insertResult(list)
    })

    session.stop()
  }

}
